package org.niit.service

import org.apache.spark.SparkConf
import org.apache.spark.sql.{DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.streaming.{OutputMode, Trigger}
import org.niit.bean.AdClickData

class WeiTimeService {
  // 初始化Spark配置（适配项目结构）
  private val sparkConf = new SparkConf()
    .setMaster("local[*]")
    .setAppName("WeiTimeService")
    .set("spark.sql.streaming.checkpointLocation", "/tmp/wei_checkpoint/")
    .set("spark.sql.session.timeZone", "Asia/Shanghai")

  private val spark = SparkSession.builder().config(sparkConf).getOrCreate()
  spark.sparkContext.setLogLevel("ERROR")
  import spark.implicits._

  /**
   * 读取Kafka流并转换为AdClickData样例类
   */
  private def getKafkaStream(): DataFrame = {
    spark.readStream
      .format("kafka")
      .option("kafka.bootstrap.servers", "localhost:9092")
      .option("subscribe", "user_behavior_topic")
      .option("failOnDataLoss", "false")
      .load()
      .selectExpr("CAST(value AS STRING)")
      .select(
        // 直接映射到AdClickData的字段（保持字符串类型，与样例类一致）
        get_json_object(col("value"), "$.time").alias("time"),
        get_json_object(col("value"), "$.user_id").alias("user_id"),
        get_json_object(col("value"), "$.new_user").alias("new_user"),
        get_json_object(col("value"), "$.age").alias("age"),
        get_json_object(col("value"), "$.sex").alias("sex"),
        get_json_object(col("value"), "$.market").alias("market"),
        get_json_object(col("value"), "$.device").alias("device"),
        get_json_object(col("value"), "$.operative_system").alias("operative_system"),
        get_json_object(col("value"), "$.source").alias("source"),
        get_json_object(col("value"), "$.total_pages_visited").alias("total_pages_visited"),
        get_json_object(col("value"), "$.home_page").alias("home_page"),
        get_json_object(col("value"), "$.listing_page").alias("listing_page"),
        get_json_object(col("value"), "$.product_page").alias("product_page"),
        get_json_object(col("value"), "$.payment_page").alias("payment_page"),
        get_json_object(col("value"), "$.payment_confirmation_page").alias("payment_confirmation_page")
      )
      .as[AdClickData] // 转换为样例类
      .withColumn("event_time", to_timestamp(col("time"))) // 转为时间戳用于窗口计算
  }

  /**
   * 启动所有实时分析任务
   */
  def startAllAnalyses(): Unit = {
    val streamData = getKafkaStream()

    // 1. 实时查看不同设备上用户访问主页的转化率波动 → output/WeiTime1.json
    analysis1(streamData)

    // 2. 实时统计各来源渠道用户浏览列表页的实时转化率 → output/WeiTime2.json
    analysis2(streamData)

    // 3. 实时查看各来源渠道用户在不同操作系统下浏览主页的实时转化率 → output/WeiTime3.json
    analysis3(streamData)

    // 4. 实时分析每半小时内不同性别用户浏览产品详情页的实时数量变化 → output/WeiTime4.json
    analysis4(streamData)

    // 5. 实时统计各来源渠道用户在不同性别下浏览主页的实时转化率 → output/WeiTime5.json
    analysis5(streamData)

    // 6. 实时跟踪新用户在不同操作系统下浏览支付页面的实时转化率 → output/WeiTime6.json
    analysis6(streamData)

    spark.streams.awaitAnyTermination()
  }

  /**
   * 1. 不同设备上用户访问主页的转化率波动
   */
  private def analysis1(streamData: DataFrame): Unit = {
    streamData
      .groupBy(
        window(col("event_time"), "5 minutes"),
        col("device")
      )
      .agg(
        countDistinct("user_id").alias("total_users"),
        sum(col("home_page").cast("int")).alias("home_visits"),
        (sum(col("home_page").cast("int")) / countDistinct("user_id")).alias("conversion_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("window.end").alias("window_end"),
        col("device"),
        col("total_users"),
        col("home_visits"),
        col("conversion_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime1.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis1/")
      .trigger(Trigger.ProcessingTime("1 minute"))
      .start()
  }

  /**
   * 2. 各来源渠道用户浏览列表页的实时转化率
   */
  private def analysis2(streamData: DataFrame): Unit = {
    streamData
      .filter(col("home_page").cast("int") === 1)
      .groupBy(
        window(col("event_time"), "10 minutes"),
        col("source")
      )
      .agg(
        countDistinct("user_id").alias("home_users"),
        sum(col("listing_page").cast("int")).alias("listing_visits"),
        (sum(col("listing_page").cast("int")) / countDistinct("user_id")).alias("conversion_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("source"),
        col("home_users"),
        col("listing_visits"),
        col("conversion_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime2.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis2/")
      .trigger(Trigger.ProcessingTime("2 minutes"))
      .start()
  }

  /**
   * 3. 各来源渠道在不同操作系统下浏览主页的实时转化率
   */
  private def analysis3(streamData: DataFrame): Unit = {
    streamData
      .groupBy(
        window(col("event_time"), "15 minutes"),
        col("source"),
        col("operative_system")
      )
      .agg(
        countDistinct("user_id").alias("total_users"),
        sum(col("home_page").cast("int")).alias("home_visits"),
        (sum(col("home_page").cast("int")) / countDistinct("user_id")).alias("conversion_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("source"),
        col("operative_system"),
        col("total_users"),
        col("conversion_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime3.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis3/")
      .trigger(Trigger.ProcessingTime("3 minutes"))
      .start()
  }

  /**
   * 4. 每半小时内不同性别用户浏览产品详情页的数量变化
   */
  private def analysis4(streamData: DataFrame): Unit = {
    streamData
      .filter(col("product_page").cast("int") === 1)
      .groupBy(
        window(col("event_time"), "30 minutes"),
        col("sex")
      )
      .agg(
        count("user_id").alias("detail_visits"),
        lag(count("user_id"), 1, 0).over(
          Window.partitionBy("sex").orderBy("window.start")
        ).alias("prev_visits"),
        when(
          col("prev_visits") === 0, 0.0
        ).otherwise(
          (count("user_id") - col("prev_visits")) / col("prev_visits")
        ).alias("mom_change_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("sex"),
        col("detail_visits"),
        col("prev_visits"),
        col("mom_change_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime4.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis4/")
      .trigger(Trigger.ProcessingTime("5 minutes"))
      .start()
  }

  /**
   * 5. 各来源渠道在不同性别下浏览主页的实时转化率
   */
  private def analysis5(streamData: DataFrame): Unit = {
    streamData
      .groupBy(
        window(col("event_time"), "10 minutes"),
        col("source"),
        col("sex")
      )
      .agg(
        countDistinct("user_id").alias("total_users"),
        sum(col("home_page").cast("int")).alias("home_visits"),
        (sum(col("home_page").cast("int")) / countDistinct("user_id")).alias("conversion_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("source"),
        col("sex"),
        col("conversion_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime5.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis5/")
      .trigger(Trigger.ProcessingTime("2 minutes"))
      .start()
  }

  /**
   * 6. 新用户在不同操作系统下浏览支付页面的实时转化率
   */
  private def analysis6(streamData: DataFrame): Unit = {
    streamData
      .filter(col("new_user").cast("int") === 1)
      .groupBy(
        window(col("event_time"), "5 minutes"),
        col("operative_system")
      )
      .agg(
        countDistinct("user_id").alias("new_total_users"),
        sum(col("payment_page").cast("int")).alias("payment_visits"),
        (sum(col("payment_page").cast("int")) / countDistinct("user_id")).alias("conversion_rate")
      )
      .select(
        col("window.start").alias("window_start"),
        col("operative_system"),
        col("new_total_users"),
        col("payment_visits"),
        col("conversion_rate")
      )
      .writeStream
      .outputMode(OutputMode.Update())
      .format("json")
      .option("path", "output/WeiTime6.json")
      .option("checkpointLocation", "/tmp/wei_checkpoint/analysis6/")
      .trigger(Trigger.ProcessingTime("1 minute"))
      .start()
  }
}